package org.example.threadpool;

import lombok.extern.slf4j.Slf4j;
import org.example.utils.Sleeper;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author zhengdp
 * @Description 自定义线程池
 * 拒绝策略 ：
 * 1. 死等
 * 2. 带超时的等待
 * 3. 让调用者放弃任务的执行
 * 4. 让调用者抛出异常
 * 5. 让调用者自己执行任务
 * @Date 2022/12/6 22:18
 * @Created by zhengdp
 */
@Slf4j(topic = "c.CustomPool")
public class CustomPool {
    public static void main(String[] args) {
        //ThreadPool pool = new ThreadPool(2,1000,TimeUnit.MICROSECONDS,10);
        //for(int i = 0; i< 15 ; i++) {
        //    int j = i;
        //    pool.execute(new Runnable() {
        //        @Override
        //        public void run() {
        //            // 阻塞添加，当任务队列满了，主线程就会卡住，等待队列有位置 ==> 定制拒绝策略
        //            Sleeper.sleep(10000);
        //            log.info("{}",j);
        //        }
        //    });
        //}

        // 带拒绝策略
        ThreadPool pool = new ThreadPool(1, 1000, TimeUnit.MICROSECONDS, 1, (queue, task) -> {
            // 1.死等
            queue.put(task);
            // 2. 超时等待
            //queue.offer(task,1500,TimeUnit.MICROSECONDS);
            // 3. 放弃任务执行
            //log.info("放弃任务 {}",task);
            // 4. 抛出异常 ==> 抛出异常，会让剩余的任务都不执行
            //  主线程调用线程池的 execute()方法，execute()方法中调用reject()方法，
            //  抛出异常，所以是调用者抛出的异常
            //throw new RuntimeException("任务执行失败 " + task);
            // 5. 调用者自己执行任务
            //task.run();
        });
        for(int i = 0; i< 3 ; i++) {
            int j = i;
            pool.execute(new Runnable() {
                @Override
                public void run() {
                    // 阻塞添加，当任务队列满了，主线程就会卡住，等待队列有位置 ==> 定制拒绝策略
                    Sleeper.sleep(5000);
                    log.info("{}",j);
                }
            });
        }
    }
}

/**
 * 拒绝策略接口
 */
@FunctionalInterface
interface  RejectPolicy<T> {
    void reject(BlockingQueue<T> queue,T task);
}

@Slf4j(topic = "c.ThreadPool")
class ThreadPool {
    // 任务队列
    private BlockingQueue<Runnable> taskQueue;

    // 线程集合 （需要包装线程）
    private HashSet<Worker> workers = new HashSet();

    // 核心线程数
    private  int coreSize;

    // 超时时间（获取任务的超时时间） ==> 一旦超过时间没有获取到任务，那么线程就会被销毁
    private long timeout;
    private TimeUnit timeUnit;

    // 拒绝策略
    private RejectPolicy<Runnable> rejectPolicy;

    public ThreadPool( int coreSize, long timeout,
                      TimeUnit timeUnit,int queueCapacity,RejectPolicy<Runnable> rejectPolicy) {
        this.taskQueue = new BlockingQueue<>(queueCapacity);
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.rejectPolicy = rejectPolicy;
    }

    /**
     * 执行任务
     * @param task 任务
     */
    public void execute (Runnable task) {
        // workers 是一个共享变量，需要保证它的线程安全
        synchronized (workers) {
            // 当任务数 没有超过 coreSize 时，直接将任务 交给 Worker对象执行
            if(workers.size() < coreSize) {
                log.info("新增 worker : {},任务 :{}",workers,task);
                Worker worker = new Worker(task);
                workers.add(worker);
                // 执行任务
                worker.start();
            }else{
                log.info("将任务:{}加入任务队列",task);
                // 如果任务数 超过 coreSize ，将任务加入 任务队列（阻塞队列）暂存
                taskQueue.tryPut(rejectPolicy,task);
            }
        }

    }

    class Worker  extends Thread{
        private Runnable task;
        public Worker(Runnable task) {
            this.task = task;
        }

        @Override
        public void run() {
            // 1. 当 task 不为空，执行任务
            // 2. 当 task 执行完毕，接着从任务队列获取任务并执行
            while( task != null || (task = taskQueue.poll(timeout,timeUnit)) != null) {
                try {
                    log.info("正在执行 :{}",task);
                    task.run();
                }catch ( Exception e) {
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            // 若退出循环，说明任务队列中已经没有任务，移除 worker对象
            synchronized (workers) {
                log.info("worker : {} 被移除",this);
                workers.remove(this);
            }
        }
    }
}



/**
 * 阻塞队列
 * @param <T>
 */
@Slf4j(topic = "c.BlockingQueue")
class BlockingQueue<T> {

    public BlockingQueue(int capacity) {
        this.capacity = capacity;
    }

    // 1. 任务队列
    private Deque<T> queue = new ArrayDeque<>();

    /**
     * 2. 锁
     * 线程池有多个任务，都要从任务队列的头部获取任务，需要保护队列头部的任务 （也需要保护队列尾的元素 ==> 多个生产者添加任务）
      */
    private ReentrantLock lock = new ReentrantLock();

    // 3. 生产者条件变量 ---> 当队列满时，生产者进入条件变量等待
    private Condition fullWaitSet = lock.newCondition();

    // 4. 消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5. 容量
    private int capacity;

    /**
     * 带超时的阻塞获取
     * @param timeout 超时时间
     * @param unit 时间单位
     * @return 未执行任务
     */
    public T poll (long timeout, TimeUnit unit) {
        lock.lock();
        try {
            // 将timeout 统一转换为 纳秒
            long nacos = unit.toNanos(timeout);
            while(queue.isEmpty()) {
                try{
                    // 等待已经超时
                    if ( nacos <= 0) {
                        return null;
                    }
                    // 剩余时间
                    nacos = emptyWaitSet.awaitNanos(nacos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 2. 队列不为空，返回队列头的元素
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞获取
     * @return
     */
    public T task () {
        lock.lock();
        try {
            // 1. 队列为空，进入阻塞等待
            while(queue.isEmpty()) {
                try{
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 2. 队列不为空，返回队列头的元素
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞添加
     * @param obj
     */
    public void put(T obj) {
        lock.lock();
        try{
            while( queue.size() == capacity) {
                try{
                    log.info("等待加入任务队列 {} ...",obj);
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 当队列不为空，将任务加入到队列中
            log.info("加入任务队列  {}",obj);
            queue.addLast(obj);
            // 唤醒消费者线程
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞添加 + 带超时时间
     * @param task 任务队列
     * @param timeout 超时时间
     * @param timeUnit 超时单位
     */
    public boolean offer(T task,long timeout,TimeUnit timeUnit) {
        lock.lock();
        try{
            long nacos = timeUnit.toNanos(timeout);
            while( queue.size() == capacity) {
                try{
                    if(nacos <= 0) {
                        // 添加失败
                        return false;
                    }
                    log.info("等待加入任务队列 {}",task);
                    nacos = fullWaitSet.awaitNanos(nacos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            // 当队列不为空，将任务加入到队列中
            log.info("加入任务队列  {}",task);
            queue.addLast(task);
            // 唤醒消费者线程
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy,T task) {
        lock.lock();
        try{
            // 1. 判断队列是否已满
            if(queue.size() == capacity){
                rejectPolicy.reject(this,task);
            }else{
                // 队列未满，将任务对象加入到队列
                log.info("加入任务队列  {}",task);
                queue.addLast(task);
                emptyWaitSet.signal();
            }
        }finally {
            lock.unlock();
        }
    }

    public int size () {
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }


}

